package com.test.study.ToolsUtils.strom;

import java.util.HashMap;
import java.util.Map;

import org.apache.log4j.Logger;
import org.apache.storm.Config;
import org.apache.storm.Constants;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Tuple;

/**
 * 对之前的Tuple 的数据进行处理
 * 
 * @author huyong
 *
 */
public class BoltClass extends BaseRichBolt {
	private static final Logger log = Logger.getLogger(BoltClass.class);
	@SuppressWarnings({ "unused", "rawtypes" })
	private Map stormConf;
	@SuppressWarnings("unused")
	private TopologyContext context;
	@SuppressWarnings("unused")
	private OutputCollector collector;

	private static final long serialVersionUID = 617145394450929666L;

	/**
	 * 相当于初始化方法，当这个组件被初始化的时候执行一次，并且只会执行一次 参数说明：参数1:Storm的配置信息 参数2:storm上下文
	 * 参数3:storm的发射器
	 */
	@SuppressWarnings("rawtypes")
	@Override
	public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
		this.stormConf = stormConf;
		this.context = context;
		this.collector = collector;
		System.out.println("Strom" + stormConf + "===" + stormConf + "=====context===" + context + "====collector===="
				+ collector);
	}

	int sum = 0;

	/**
	 * 死循环 一直执行
	 */
	@Override
	public void execute(Tuple input) {
		try {
			// 判断这次的消息是否是系统级别的消息 这个是定时任务的
			if (input.getSourceComponent().equals(Constants.SYSTEM_COMPONENT_ID)) {
				log.info("sum:" + sum );
				System.out.println("sum:" + sum );
				log.info("累加求和成功调用成功的方法："); // 默认是消息确认机制是没有开启的 如果没有开启这样调用是会报错的
			}else {
				// 累加求和
				Integer num = input.getIntegerByField("number");// 通过之前指定的字段获取tuple发送的数据
				System.out.println(Thread.currentThread().getId() + "\t" + num);
				sum += num;
				this.collector.ack(input);
			}
		} catch (Exception e) {
			log.error("累加求和失败调用失败的方法", e);
			this.collector.fail(input);
		}

	}

	/**
	 * 声明输出端 注意：在nexttuple或者execute方法中如果没有向外发射tuple，那么在这就不需要声明输出字段
	 */
	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		// TODO Auto-generated method stub

	}

	/**
	 * 这个方法表示只是给 这一个组建设置定时任务
	 */
	@Override
	public Map<String, Object> getComponentConfiguration() {
		HashMap<String,Object> hashMap=new HashMap<String,Object>();
		hashMap.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS, 10);//10秒
		return hashMap;
	}

}
